package com.pangyangyang.elasticsearch.plugin.river;

import com.pangyangyang.elasticsearch.plugin.util.MessageRiverConstants;
import com.pangyangyang.elasticsearch.plugin.collector.SingleCollectorManager;
import com.pangyangyang.elasticsearch.plugin.collector.EventCollector;
import com.pangyangyang.elasticsearch.plugin.dispatch.EventDispatcherHolder;
import com.pangyangyang.elasticsearch.plugin.exception.MessageRiverException;
import com.pangyangyang.elasticsearch.plugin.handler.EventHandler;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.ESLoggerFactory;
import org.elasticsearch.river.AbstractRiverComponent;
import org.elasticsearch.river.River;
import org.elasticsearch.river.RiverName;
import org.elasticsearch.river.RiverSettings;

import java.lang.reflect.Constructor;
import java.util.Map;

/**
 * Created by yangwang on 2015/1/14.
 */
public class MessageRiver extends AbstractRiverComponent implements River {

    private Client client;
    private EventCollector eventCollector;
    private EventHandler handler;
//    private MessageRiverSettings riverSettings;
    private SettingsParser parser;


    private static ESLogger logger = ESLoggerFactory.getLogger(Build.LOGGER_NAME);

    @Inject
    protected MessageRiver(RiverName riverName, RiverSettings settings, Client client) {
        super(riverName, settings);
        this.client = client;
        parser = new SettingsParser(this.settings.settings());
    }

    @Override
    public void start() {
        if (logger.isDebugEnabled()){
            printSettingDebugInfo();
        }

        String collectorString = parser.getCollectorSettings().getString(MessageRiverConstants.SETTING_KEY_COMMON_CLASS);
        String handlerString =  parser.getHandlerSettings().getString(MessageRiverConstants.SETTING_KEY_COMMON_CLASS);

        initEventCollector(collectorString);
        initEventHandler(handlerString);

        EventDispatcherHolder.registerRiver(this);
    }

    @Override
    public void close() {
        SingleCollectorManager.stopCanalCollector(this.eventCollector);
        this.handler.stop();
        EventDispatcherHolder.unregisterRiver(this);
        logger.info("message river closed.");
    }
    protected void initEventCollector(String collectorName){

        eventCollector = SingleCollectorManager.startCanalCollector(this.parser.getCollectorSettings(), collectorName);
    }

    protected void initEventHandler(String handlerName){
        logger.debug("starting handler {}", handlerName);
        Class<? extends EventHandler> clazz = Build.getInstance().getHandlerClasses().get(handlerName);
        if (clazz == null){
            throw new MessageRiverException(String.format("handler not found %s.", handlerName));
        }

        try {
            Constructor<? extends  EventHandler> constructor = clazz.getConstructor(Client.class, MessageRiverSettings.class);
            handler = constructor.newInstance(client, this.parser.getHandlerSettings());
            handler.start();
        } catch (Exception e) {
            logger.error("create handler failed {}.", e, clazz.getName());
            throw new MessageRiverException(String.format("create handler failed %s.", clazz.getName()), e);
        }
    }

    public EventHandler getHandler() {
        return handler;
    }

    public String getDestination() {
        return getHandler().getDestination();
    }

    public void printSettingDebugInfo(){
        String destination =  this.parser.getHandlerSettings().getString(MessageRiverConstants.SETTING_KEY_COMMON_DESTINATION);
        String collectorString = parser.getCollectorSettings().getString(MessageRiverConstants.SETTING_KEY_COMMON_CLASS);
        String handlerString =  parser.getHandlerSettings().getString(MessageRiverConstants.SETTING_KEY_COMMON_CLASS);
        logger.debug("start message river instance  des {},  collector  {}, handler {}.", destination, collectorString, handlerString);
    }
}
